package org.ws.task.zookeeper.node;

import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.nodes.PersistentNode;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.Bootable;
import org.ws.task.common.IPUtils;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.NodeRole;
import org.ws.task.model.NodeStatus;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.zookeeper.NodePaths;
import org.ws.task.zookeeper.listener.NodeListenerManager;

import java.io.IOException;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class WorkerNode extends AbstractNode implements Bootable {

    private static Logger log = LoggerFactory.getLogger(WorkerNode.class.getName());

    private final CuratorCache workerNodeCache;
    private final PersistentNode workerPersistentNode;
    private final AtomicReference<NodeMetadata> currentNodeMetadata;

    public WorkerNode(String id, String namespace, CuratorFramework curatorClient, NodeListenerManager nodeListenerManager) {
        super(id, namespace, curatorClient, nodeListenerManager);
        this.workerNodeCache = CuratorCache.build(this.curatorClient, NodePaths.getWorkersPath(namespace));
        this.workerNodeCache.listenable().addListener(buildWorkerNodeListener());
        NodeMetadata nodeMetadata = buildNodeData();
        this.workerPersistentNode = new PersistentNode(this.curatorClient, CreateMode.EPHEMERAL_SEQUENTIAL,true,
                NodePaths.getWorkersNodePath(namespace),NodeDataCodec.encode(nodeMetadata));
        this.currentNodeMetadata = new AtomicReference<>(nodeMetadata);
        NodeMetadataRepository.getInstance().setTheWorkerNode(nodeMetadata);
    }

    @Override
    public void start() {
        this.workerNodeCache.start();
        this.workerPersistentNode.start();
    }

    @Override
    public void shutdown() {
        this.workerNodeCache.close();
        try {
            this.workerPersistentNode.close();
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
    }

    public void updateWorker(int port, NodeStatus status){
        NodeMetadata nodeMetadata = this.currentNodeMetadata.get();
        nodeMetadata.setStatus(status);
        nodeMetadata.setPort(port);
        nodeMetadata.setUpdateTime(new Date());
        this.nodeListenerManager.updateWorkerNode(nodeMetadata);
        try {
            // 等待30s初始化
            this.workerPersistentNode.waitForInitialCreate(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error(e.getMessage(),e);
            Thread.currentThread().interrupt();
        }
        String path = this.workerPersistentNode.getActualPath();
        try {
            this.updateNodeData(path,nodeMetadata, CreateMode.EPHEMERAL_SEQUENTIAL);
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }
    }

    public Collection<NodeMetadata> getWorkers(){
        List<NodeMetadata> list = Lists.newArrayList();
        this.workerNodeCache.stream().forEach(childData -> {
            byte[] data = childData.getData();
            NodeMetadata nodeMetadata = NodeDataCodec.decode(data,NodeMetadata.class);
            if(nodeMetadata != null){
                list.add(nodeMetadata);
            }
        });
        return list;
    }

    private CuratorCacheListener buildWorkerNodeListener(){
        return CuratorCacheListener.builder()
                .afterInitialized()
                .forCreatesAndChanges((oldNode,child) -> {
                    if(oldNode == null || child == null){
                        return;
                    }
                    if(child != null){
                        byte[] data = child.getData();
                        NodeMetadata nodeMetadata = NodeDataCodec.decode(data,NodeMetadata.class);
                        if(nodeMetadata != null){
                            log.debug("update node -> {}",nodeMetadata.toString());
                            WorkerNode.this.nodeListenerManager.updateWorkerNode(nodeMetadata);
                        }
                    }
                })
                .forDeletes(childData -> {
                    byte[] data = childData.getData();
                    if(data == null || data.length <= 0){
                        return;
                    }
                    // 删除节点
                    NodeMetadata nodeMetadata = NodeDataCodec.decode(data,NodeMetadata.class);
                    log.debug("delete node -> {}",nodeMetadata);
                    if(nodeMetadata != null){
                        WorkerNode.this.nodeListenerManager.removeWorkerNode(nodeMetadata);
                    }
                })
                .build();
    }

    private NodeMetadata buildNodeData(){
        return NodeMetadata.builder()
                .id(this.id)
                .host(IPUtils.getIP())
                .port(-1)
                .role(NodeRole.WORKER)
                .status(NodeStatus.READY)
                .updateTime(new Date())
                .build();
    }
}
